package com.huawei.bigdata.kafka.temp;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.RecordDeserializationException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;
import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
import org.apache.kafka.common.KafkaException;

@Slf4j
@Service
public class TestConsumer {
//    @Value("${conf.kafka.test.topic}")
//    private String topic;
    @Value("${conf.kafka.test.bootstrap-servers}")
    private String bootstrapServers;
    @Value("${conf.kafka.test.security-protocol}")
    private String securityProtocol;
    @Value("${conf.kafka.test.kerberos-domain-name}")
    private String kerberosDomainName;
    @Value("${conf.kafka.test.group-id}")
    private String groupId;
    @Value("${conf.kafka.test.enable-auto-commit}")
    private String enableAutoCommit;
    @Value("${conf.kafka.test.security}")
    private Boolean security;

    private volatile boolean closed;

    // 一次请求的最大等待时间(S)
    private final int waitTime = 1;

    // Broker连接地址
    private final static String BOOTSTRAP_SERVER = "bootstrap.servers";

    // Group id
    private final static String GROUP_ID = "group.id";

    // 消息内容使用的反序列化类
    private final static String VALUE_DESERIALIZER = "value.deserializer";

    // 消息Key值使用的反序列化类
    private final static String KEY_DESERIALIZER = "key.deserializer";

    // 协议类型:当前支持配置为SASL_PLAINTEXT或者PLAINTEXT
    private final static String SECURITY_PROTOCOL = "security.protocol";

    // 服务名
    private final static String SASL_KERBEROS_SERVICE_NAME = "sasl.kerberos.service.name";

    // 域名
    private final static String KERBEROS_DOMAIN_NAME = "kerberos.domain.name";

    // 是否自动提交offset
    private final static String ENABLE_AUTO_COMMIT = "enable.auto.commit";

    // 自动提交offset的时间间隔
    private final static String AUTO_COMMIT_INTERVAL_MS = "auto.commit.interval.ms";

    // 会话超时时间
    private final static String SESSION_TIMEOUT_MS = "session.timeout.ms";
    public Properties initProperties() {
        Properties props = new Properties();

        // Broker连接地址
        props.put(BOOTSTRAP_SERVER, bootstrapServers);
        // Group id
        props.put(GROUP_ID, groupId);
        // 是否自动提交offset
        props.put(ENABLE_AUTO_COMMIT, enableAutoCommit);
        // 自动提交offset的时间间隔
        props.put(AUTO_COMMIT_INTERVAL_MS, "1000");
        // 会话超时时间
        props.put(SESSION_TIMEOUT_MS, "30000");
        // 消息Key值使用的反序列化类
        props.put(KEY_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer");
        // 消息内容使用的反序列化类
        props.put(VALUE_DESERIALIZER, "org.apache.kafka.common.serialization.StringDeserializer");
        if (security) {
            // 安全协议类型
            props.put(SECURITY_PROTOCOL, securityProtocol);
        }
        // 服务名
        props.put(SASL_KERBEROS_SERVICE_NAME, "kafka");
        // 域名
        props.put(KERBEROS_DOMAIN_NAME, kerberosDomainName);

        return props;
    }
    private boolean running = false;
    public void consume(String topic) {
        if (running) return;
        running = true;
        Properties props = initProperties();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        // 订阅
        consumer.subscribe(Collections.singletonList(topic));
        while (!closed) {
            try {
                // 消息消费请求
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(waitTime));
                // 消息处理
                for (ConsumerRecord<String, String> record : records) {
                    log.info("topicname: "+topic+", Received message: (" + record.key() + ", " + record.value()
                            + ") at offset " + record.offset());
                }
            } catch (AuthorizationException | UnsupportedVersionException
                     | RecordDeserializationException e) {
                log.error(e.getMessage());
                // 无法从异常中恢复
                closeThread();
                latchShutDown();
            } catch (OffsetOutOfRangeException | NoOffsetForPartitionException e) {
                log.error("Invalid or no offset found, using latest");
                consumer.seekToEnd(e.partitions());
                consumer.commitSync();
            } catch (KafkaException e) {
                log.error(e.getMessage());
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
    public void closeThread() {
        if (!closed) {
            closed = true;
        }
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        this.running = false;
    }

    public void latchShutDown() {
        new CountDownLatch(1).countDown();
    }

}
